package com.ztorn.driver;

import com.ztorn.cdc.CDCBuilder;
import com.ztorn.common.assertion.Asserts;
import com.ztorn.common.model.FlinkCDCConfig;
import com.ztorn.common.model.Schema;
import com.ztorn.common.model.Table;
import com.ztorn.common.util.SqlUtil;
import com.ztorn.framework.driver.Driver;
import com.ztorn.framework.driver.DriverConfig;

import java.util.*;

public class DriverSchema {

    public static Driver checkAndCreateSinkSchema(Map<String, String> sink, String schemaName) throws Exception {
        String autoCreate = sink.get(FlinkCDCConfig.AUTO_CREATE);
        if (!Asserts.isEqualsIgnoreCase(autoCreate, "true") || Asserts.isNullString(schemaName)) {
            return null;
        }
        String url = sink.get("url");
        String schema = SqlUtil.replaceAllParam(sink.get(FlinkCDCConfig.SINK_DB), "schemaName", schemaName);
        Driver driver = Driver.build(sink.get("sink-connector"), url, sink.get("username"), sink.get("password"));
        if (null != driver && !driver.existSchema(schema)) {
            driver.createSchema(schema);
        }
//        sink.put(FlinkCDCConfig.SINK_DB, schema);
        return driver;
    }

    public static List<String> getSchemaTableNameList(List<Schema> schemaList, FlinkCDCConfig config, CDCBuilder cdcBuilder) throws Exception {
        final List<String> schemaNameList = cdcBuilder.getSchemaList();
        final List<String> tableRegList = cdcBuilder.getTableList();
        final List<String> schemaTableNameList = new ArrayList<>();
        Map<String, String> splitConfig = config.getSplit();
        DriverConfig driverConfig = DriverConfig.build(cdcBuilder.parseMetaDataConfig());
        Driver driver = Driver.build(driverConfig);
        Map<String, Map<String, String>> allConfigMap = cdcBuilder.parseMetaDataConfigs();
        Set<Table> tables = driver.getSplitTables(tableRegList, splitConfig);
        for(Table table:tables){
            String schemaName = table.getSchema();
            Schema schema = Schema.build(schemaName);
            schema.setTables(Collections.singletonList(table));
            // 分库分表所有表结构都是一样的，取出列表中第一个表名即可
            String schemaTableName = table.getSchemaTableNameList().get(0);
            // 真实的表名
            String tableName = schemaTableName.split("\\.")[1];
            table.setColumns(driver.listColumnsSortByPK(schemaName, tableName));
            table.setColumns(driver.listColumnsSortByPK(schemaName, table.getName()));
            schemaList.add(schema);
        }
        for(String schemaName:schemaNameList){
            Schema schema = Schema.build(schemaName);
            if (!allConfigMap.containsKey(schemaName)) {
                continue;
            }
            Map<String, String> sink = config.getSink();
            Driver sinkDriver = DriverSchema.checkAndCreateSinkSchema(sink, schemaName);

            DriverConfig driverConfig1 = DriverConfig.build(allConfigMap.get(schemaName));
            Driver driver1 = Driver.build(driverConfig1);
            final List<Table> tables1 = driver1.listTables(schemaName);
            for (Table table : tables1) {
                if (!Asserts.isEquals(table.getType(), "VIEW")) {
                    if (Asserts.isNotNullCollection(tableRegList)) {
                        for (String tableReg : tableRegList) {
                            if (table.getSchemaTableName().matches(tableReg.trim())
                                    && !schema.getTables()
                                    .contains(Table.build(table.getName()))) {
                                table.setColumns(
                                        driver.listColumnsSortByPK(
                                                schemaName, table.getName()));
                                schema.getTables().add(table);
                                schemaTableNameList.add(table.getSchemaTableName());
                                break;
                            }
                        }
                    } else {
                        table.setColumns(
                                driver.listColumnsSortByPK(schemaName, table.getName()));
                        schemaTableNameList.add(table.getSchemaTableName());
                        schema.getTables().add(table);
                    }
                }
            }
            if (null != sinkDriver) {
                for (Table table : schema.getTables()) {
                    Table sinkTable = (Table) table.clone();
                    sinkTable.setSchema(table.getSchema());
                    sinkTable.setName(table.getName());
                    checkAndCreateSinkTable(sinkDriver, sinkTable);
                }
            }
            schemaList.add(schema);
        }
        return schemaTableNameList;
    }

    static void checkAndCreateSinkTable(Driver driver, Table table) throws Exception {
        if (null != driver && !driver.existTable(table)) {
            driver.generateCreateTable(table);
        }
    }
}
